package com.tensai.study.java.newconcept.typereactiveprogram.reactor;

import org.junit.Test;
import reactor.core.publisher.*;

import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

/**
 * Processor sink 可以持续发布
 *
 * @author Tensai
 */
public class ProcessorTest {

	/**
	 * 支持多个Subscriber
	 */
	@Test
	public void directProcessorTest() {
		DirectProcessor<Integer> directProcessor = DirectProcessor.create();
		Flux<Integer> flux = directProcessor
				.filter(e -> e % 2 == 0);

		flux.subscribe(System.err::println);
		flux.subscribe(System.err::println);

		IntStream.range(1, 10)
				.forEach(directProcessor::onNext);

	}

	/**
	 * 只能有一个subscriber
	 * sink.next 之后无影响
	 *
	 * @throws InterruptedException 异常
	 */
	@Test
	public void unicastProcessorTest() throws InterruptedException {
		UnicastProcessor<Integer> unicastProcessor
				= UnicastProcessor.create();

		FluxSink<Integer> sink = unicastProcessor.sink(FluxSink.OverflowStrategy.BUFFER);

		Flux<Integer> flux = unicastProcessor
				.filter(e -> e % 2 == 0)
				.doOnError(System.err::println);

		//System.err.println("emmit data " + e);
		IntStream.range(1, 20)
				.forEach(sink::next);


		Thread.sleep(1000);
		flux.subscribe(System.err::println);
//		flux.subscribe(System.err::println);
		unicastProcessor.onComplete();

	}

	/**
	 * processor.onNext 之后订阅无效
	 */
	@Test
	public void emitterProcessorTest() {
		EmitterProcessor<Integer> emitterProcessor
				= EmitterProcessor.create(4);

		Flux<Integer> flux1 = emitterProcessor.filter(e -> e % 2 == 0);
		Flux<Integer> flux2 = emitterProcessor.filter(e -> e % 3 == 0);

		System.err.println("start subscribe ");

		flux1.subscribe(System.err::println);

		IntStream.rangeClosed(1, 10)
				.forEach(e -> {
					System.err.println("emmit data " + e);
					emitterProcessor.onNext(e);
					try {
						TimeUnit.MILLISECONDS.sleep(10);
					} catch (InterruptedException e1) {
						e1.printStackTrace();
					}
				});


		//flux2 没有消费1~10,只能消费后面的3
		flux2.subscribe(System.err::println);
		emitterProcessor.onNext(3);
		emitterProcessor.onComplete();

	}
}
